package com.pw.study.flink.demo

import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable

/**
 * @Desc: TODO
 * @author: future
 * @since: 2022/5/15-14:37
 */
object OrderEventDemo {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(3)
    val dsOrder: DataStream[String] = env.readTextFile("data/file/OrderLog.csv")
    val dsReceipt: DataStream[String] = env.readTextFile("data/file/ReceiptLog.csv")

    val dsOrderEvent = dsOrder.map(bean => {
      val line: Array[String] = bean.split(",")
      OrderEvent(line(0), line(1), line(2), line(3))
    }).keyBy(_.txId)
    val dsTx = dsReceipt.map(bean => {
      val line: Array[String] = bean.split(",")
      TxEvent(line(0), line(1), line(2))
    }).keyBy(_.txId)

    dsOrderEvent.connect(dsTx).process(new CoProcessFunction[OrderEvent, TxEvent, String] {
      val orderMap = mutable.Map[String, String]()
      val txMap = mutable.Map[String, String]()

      override def processElement1(order: OrderEvent, context: CoProcessFunction[OrderEvent, TxEvent, String]#Context, out: Collector[String]): Unit = {
        val txId: Option[String] = txMap.get(order.txId)
        if (txId.isEmpty) {
          orderMap.put(order.txId, order.orderId)
        } else {
          out.collect("订单：" + order.orderId + " 对账成功，交易ID： " + order.txId)
          txMap.remove(order.txId)
        }
      }

      override def processElement2(tx: TxEvent, context: CoProcessFunction[OrderEvent, TxEvent, String]#Context, collector: Collector[String]): Unit = {
        val txId: Option[String] = orderMap.get(tx.txId)
        if (txId.isEmpty) {
          txMap.put(tx.txId,"1")
        }else{
          collector.collect("订单："+txId.get+" 对账成功，交易ID： "+tx.txId)
          orderMap.remove(tx.txId)
        }
      }
    })

    dsOrderEvent.print()
    dsTx.print()

    env.execute("order")
  }
}

case class OrderEvent(orderId: String, eventType: String, txId: String, ts: String)

case class TxEvent(txId: String, eventType: String, ts: String)
